package com.gopay.common.dss.factory;

import com.gopay.common.constants.dic.GopayGlobalVariablesConstant;
import com.gopay.common.dss.DssException;
import com.gopay.common.dss.config.DssConfig;
import com.gopay.common.dss.tools.FuseTool;
import com.gopay.common.util.spring.SpringContextUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.CloseableUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Component;

/**
 * Created by zhaojt on 2017/7/28.
 *
 *  Curator相关对象 创建工厂
 */
@Component("curatorObjectFactory")
@Lazy
public class CuratorObjectFactory  implements InitializingBean{

    private CuratorFramework curatorFramework;

    private Logger logger = LoggerFactory.getLogger(CuratorObjectFactory.class);

    @Autowired
    private DssConfig dssConfig;


    public CuratorFramework getCuratorFramework(){
        if(curatorFramework == null){
            synchronized (this){
                if(curatorFramework == null){
                    CuratorFramework tmp = CuratorFrameworkFactory.newClient(dssConfig.getZkLockConnectionStr(),
                            new ExponentialBackoffRetry(dssConfig.getZkLockBasesleeptimems(), dssConfig.getZkLockMaxretries()));
                    tmp.getConnectionStateListenable().addListener(getListener());
                    tmp.start();

                    this.curatorFramework = tmp;
                }
            }
        }
        return curatorFramework;
    }

    private ConnectionStateListener getListener() {
        return new ConnectionStateListener() {
            @Override
            public void stateChanged(CuratorFramework client, ConnectionState newState) {
                logger.info("[stateChanged]newState={}", newState);
                if ((newState == ConnectionState.SUSPENDED) || (newState == ConnectionState.LOST)) {
                    FuseTool fuseTool = (FuseTool) SpringContextUtil.getBean("fuseTool");
                    boolean fuse = fuseTool.fuse(GopayGlobalVariablesConstant.ZK_LOCK_SWITCH, GopayGlobalVariablesConstant.ZK_LOCK_FUSE_MOBILE, 1);

                    if(fuse){
                        logger.error("ConnectionStateListener 处熔断， 关闭CuratorFramework !");
                        closeCuratorFramework();
                    }
                }
            }
        };
    }


    @Override
    public void afterPropertiesSet() throws Exception {
        //配置检测
        checkConfig();
    }

    private void checkConfig() throws DssException {
        if(dssConfig == null){
            throw new DssException("DssConfig 对象未完成初始化 !");
        }
        if(StringUtils.isBlank(dssConfig.getZkLockConnectionStr())){
            throw new DssException("zookeeper 分布式锁集群节点路径未配置 !");
        }
    }

    public void closeCuratorFramework(){
        synchronized (this) {
            CloseableUtils.closeQuietly(curatorFramework);
            curatorFramework = null;
        }
    }


}
